package org.ws.task.netty.support;

import com.google.common.util.concurrent.Uninterruptibles;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.event.support.startup.BootstrapEvent;
import org.ws.task.event.support.startup.BootstrapEventType;
import org.ws.task.event.support.BootstrapEventHolder;
import org.ws.task.common.AddressUtils;
import org.ws.task.exception.ClientStartFailedException;
import org.ws.task.netty.handler.ActionHandlerRegistry;
import org.ws.task.netty.NettyConfig;
import org.ws.task.netty.NettyServer;
import org.ws.task.netty.codec.MessageDecoder;
import org.ws.task.netty.codec.MessageEncoder;
import org.ws.task.netty.manager.ChannelConnectManager;
import org.ws.task.netty.manager.support.DefaultChannelConnectManager;
import org.ws.task.model.protocol.base.RequestMessage;
import org.ws.task.model.protocol.base.ResponseMessage;

import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

public class DefaultNettyServer extends AbstractNettyService implements NettyServer {

    private static Logger log = LoggerFactory.getLogger(DefaultNettyServer.class.getName());

    private final ServerBootstrap serverBootstrap;
    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;
    private final CountDownLatch startCountDownLatch;

    private final ChannelConnectManager channelConnectManager;
    private final AtomicBoolean running = new AtomicBoolean(false);
    private final AtomicBoolean active = new AtomicBoolean(false);
    private final ExecutorService callbackExecutor;
    private volatile int port = -1;

    public DefaultNettyServer(NettyConfig nettyConfig){
        super(nettyConfig);
        this.startCountDownLatch = new CountDownLatch(1);
        this.serverBootstrap = new ServerBootstrap();
        this.bossGroup = new NioEventLoopGroup(1,new DefaultThreadFactory("NettyServerBoss"));
        this.workerGroup = new NioEventLoopGroup(this.nettyConfig.getWorkerGroupThreadNum(),new DefaultThreadFactory("NettyServerWorker"));

        this.channelConnectManager = new DefaultChannelConnectManager();

        this.callbackExecutor = Executors.newFixedThreadPool(this.nettyConfig.getExecutorThreadNum(),new DefaultThreadFactory("NettyServerCallback"));
    }

    @Override
    public void start() {
        if(this.running.get()){
            return;
        }
        this.running.set(true);
        this.serverBootstrap.group(this.bossGroup,this.workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, SO_SNDBUF)
                .childOption(ChannelOption.SO_RCVBUF, SO_RCVBUF)
                .localAddress(new InetSocketAddress(this.nettyConfig.getServerPort()))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new LoggingHandler(LogLevel.DEBUG))
                                .addLast(new MessageDecoder())
                                .addLast(new MessageEncoder())
                                .addLast(new IdleStateHandler(0,0,DefaultNettyServer.this.nettyConfig.getIdleTimeoutMillis()))
                                .addLast(new ChannelConnectManageHandler())
                                .addLast(new ResponseMessageHandler())
                                .addLast(new RequestMessageHandler());
                    }
                });

        try {
            ChannelFuture channelFuture = this.serverBootstrap.bind().sync();
            InetSocketAddress localAddress = (InetSocketAddress)channelFuture.channel().localAddress();
            this.port = localAddress.getPort();
            this.active.set(this.port == this.nettyConfig.getServerPort());
            log.info("server start:{}",AddressUtils.getStringSocketAddress(localAddress));
        } catch (Exception e) {
            throw new ClientStartFailedException(e.getMessage(),e);
        }
        finally {
            this.startCountDownLatch.countDown();
            BootstrapEventHolder.getInstance()
                    .publishEvent(new BootstrapEvent(this.nettyConfig.getServerPort(), BootstrapEventType.SERVER_STARTUP));
        }
        log.info("netty server start.");
    }

    @Override
    public int port() {
        return this.port;
    }


    @Override
    public boolean isActive() {
        return this.active.get();
    }

    @Override
    public void shutdown() {
        try {
            this.bossGroup.shutdownGracefully();
            this.workerGroup.shutdownGracefully();
        } catch (Exception e) {
            log.error("NettyRemotingServer shutdown exception, ", e);
        }

        if (this.callbackExecutor != null) {
            try {
                this.callbackExecutor.shutdown();
            } catch (Exception e) {
                log.error("NettyRemotingServer shutdown exception, ", e);
            }
        }
        this.running.set(false);
    }

    @Override
    public boolean awaitStart(long timeout) {
        return Uninterruptibles.awaitUninterruptibly(this.startCountDownLatch,timeout, TimeUnit.MILLISECONDS);
    }

    @Override
    public ChannelConnectManager getChannelConnectManager() {
        return this.channelConnectManager;
    }

    @Override
    public void close() throws Exception {
        shutdown();
    }

    @Override
    protected ExecutorService getCallbackExecutor() {
        return this.callbackExecutor;
    }


    @Override
    public void actionHandlerRegistry(ActionHandlerRegistry actionHandlerRegistry, ExecutorService executor) {
        setActionHandlerRegistry(actionHandlerRegistry,executor);
    }

    @Override
    public ResponseMessage sync(Channel channel, RequestMessage request, long timeout) throws Exception {
        return doSync(channel, request, timeout);
    }

    @Override
    public void async(Channel channel, RequestMessage request, long timeout, Consumer<ResponseFuture> callback) throws Exception {
        doAsync(channel, request, timeout, callback);
    }

    @Override
    public void oneWay(Channel channel, RequestMessage request, long timeout) throws Exception {
        doOneWay(channel,request,timeout);
    }

    class ChannelConnectManageHandler extends ChannelDuplexHandler {
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            String remote = AddressUtils.getChannelRemoteAddress(channel);
            super.channelRegistered(ctx);
            log.info("channel Registered remote:{}",remote);
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            String remote = AddressUtils.getChannelRemoteAddress(channel);
            super.channelUnregistered(ctx);
            log.info("channel Unregistered remote:{}",remote);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            String remote = AddressUtils.getChannelRemoteAddress(channel);
            super.channelActive(ctx);
            DefaultNettyServer.this.channelConnectManager.onConnect(remote,channel);
            BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(remote, BootstrapEventType.CHANNEL_CONNECT));
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            String remote = AddressUtils.getChannelRemoteAddress(channel);
            super.channelInactive(ctx);
            DefaultNettyServer.this.channelConnectManager.onClose(remote,channel);
            BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(remote, BootstrapEventType.CHANNEL_CLOSE));
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            Channel channel = ctx.channel();
            String remote = AddressUtils.getChannelRemoteAddress(channel);
            super.exceptionCaught(ctx, cause);
            DefaultNettyServer.this.channelConnectManager.onException(remote,channel);
            BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(remote, BootstrapEventType.CHANNEL_CLOSE));
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if(evt instanceof IdleStateEvent){
                IdleStateEvent event = (IdleStateEvent) evt;
                if(event.state().equals(IdleState.ALL_IDLE)){
                    Channel channel = ctx.channel();
                    final String remote = AddressUtils.getChannelRemoteAddress(channel);
                    log.warn("netty client pipeline: idle exception {}", remote);
                    DefaultNettyServer.this.channelConnectManager.onIdle(remote,channel);
                    BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(remote, BootstrapEventType.CHANNEL_IDLE));
                }
            }
            super.userEventTriggered(ctx, evt);
        }
    }

}
